package com.synthefact.daj.model.threaded.impl;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.synthefact.daj.model.ChannelIf;
import com.synthefact.daj.model.MessageIf;
import com.synthefact.daj.model.NodeIf;

/**
 * A channel is a communication link between a sender process and a receiver
 * process. A channel buffers any number of messages from the sender process.
 * <p>
 * Copyright (C) 2011 Synthefact
 * <p>
 * Based on code from (c) 1997, Wolfgang Schreiner <Wolfgang.Schreiner@risc.uni-linz.ac.at>
 * http://www.risc.uni-linz.ac.at/software/daj
 */
public class Channel implements ChannelIf {

    final Logger logger = LoggerFactory.getLogger(Channel.class);

    /** message queue */
    // private MessageQueue queue = new MessageQueue();

    /** time to wait on a poll request of the channel */
    private static final long POLL_TIME_OUT = 350;

    /** The node that pushes messages to this channel */
    private NodeIf iSender;

    /** The node that pulls messages from this channel */
    private NodeIf iReceiver;

    /**
     * the queue used to communicate between the sender and receiver on this
     * channel
     */
    private LinkedBlockingQueue iQueue;

    /**
     * Construct a new channel with a sender and a receiver node
     * 
     * @param sender
     *            the node that pushes messages to this channel
     * @param receiver
     *            the node that pulls messages from this channel
     */
    public Channel(NodeIf sender, NodeIf receiver) {
	iSender = sender;
	iReceiver = receiver;
	iQueue = new LinkedBlockingQueue<MessageIf>();
    }

    @Override
    public NodeIf getSender() {
	return iSender;
    }

    @Override
    public NodeIf getReceiver() {
	return iReceiver;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void send(MessageIf msg) {
	iQueue.add(msg);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public MessageIf receive() {
	return (MessageIf) iQueue.remove();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public MessageIf receive(int count) {
	MessageIf retValue = null;
	try {

	    for (int i = 0; i < count; i++) {
		retValue = (MessageIf) iQueue.poll(POLL_TIME_OUT, TimeUnit.MILLISECONDS);
		if (retValue != null) {
		    break;
		}
	    }
	    return retValue;
	} catch (InterruptedException e) {
	    logger.debug(String.format("Interrupted: %s", e.getMessage()));
	}
	return retValue;
    }

    public List<MessageIf> getMessages() {
	return new ArrayList<MessageIf>(iQueue);
    }

    @Override
    public boolean isReady() {
	return (iQueue.size() > 0);
    }

    public String toString() {
	return String.format("Sender: %s, Receiver: %s, Count: %d", iSender, iReceiver, iQueue.size());
    }
}
